package dataStream;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/8/5
 * @Update 10:33
 */
@Slf4j
public class KafkaSinkBasic {
    public static final String BOOTSTRAPS = "172.17.1.101:9092,172.17.1.102:9092,172.17.1.103:9092";
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", BOOTSTRAPS);
        properties.setProperty("group.id", "test");


        FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<>(
                "test_20210805",
                new SimpleStringSchema(),
                properties);

        DataStream<String> stream = env.readTextFile("E:\\test_data\\kafkaSink.txt");
        stream.addSink(myProducer);
        log.info("zhixingjihua");
        log.info(env.getExecutionPlan());
        env.execute();

    }
}
